package com.atgugu.flink.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneOffset;

/**
 * @Author lzc
 * @Date 2022/4/11 9:12
 */
public class Flink15_Window_Group_1 {
    public static void main(String[] args) {
        // 流->动态表->连续查询->动态表->流
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));
        
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int," +
                            " et as to_timestamp_ltz(ts, 3), " +
                            " watermark for et as et - interval '3' second" +  // 时间属性, 类型必须是 timestamp(3)  bigint  ->   timestamp(3)
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/sensor.txt', " +
                            " 'format' = 'csv' " +
                            ")");
        
        //  select id, w, sum(vc) from t group by id, w
       /* tEnv
            .sqlQuery("select " +
                          " id, " +
                          " sum(vc) vc_sum, " +
                          " tumble_start(et, interval '5' second) stt," +
                          " tumble_end(et, interval '5' second) edt " +
                          "from sensor " +
                          "group by id, tumble(et, interval '5' second)")
            .execute()
            .print();*/
    
    
        /*tEnv
            .sqlQuery("select " +
                          " id, " +
                          " sum(vc) vc_sum, " +
                          " hop_start(et, interval '2' second, interval '5' second) stt," +
                          " hop_end(et, interval '2' second, interval '5' second) edt " +
                          "from sensor " +
                          "group by id, hop(et, interval '2' second, interval '5' second)")
            .execute()
            .print();
       */
    
    
        tEnv
            .sqlQuery("select " +
                          " id, " +
                          " sum(vc) vc_sum, " +
                          " session_start(et, interval '2' second) stt," +
                          " session_end(et, interval '2' second) edt " +
                          "from sensor " +
                          "group by id, session(et, interval '2' second)")
            .execute()
            .print();
    
    }
}
